/*
 * @Version: 0.0.1
 * @Author: ider
 * @Date: 2021-05-17 13:21:35
 * @LastEditors: ider
 * @LastEditTime: 2021-05-20 15:14:48
 * @Description: 版本控制模块， 通过消息队列发布回收任务
 */
package reversion

import (
	"encoding/json"
	"fmt"
	"historyOfWikipedia/internal/compress"
	"historyOfWikipedia/internal/config"
	"historyOfWikipedia/internal/database/mongodb"
	"historyOfWikipedia/internal/model"
	"historyOfWikipedia/pb/reversion_pb"
	"io/ioutil"
	"net/http"
	"strconv"
	"strings"
	"time"

	log "github.com/sirupsen/logrus"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/timestamppb"

	"github.com/streadway/amqp"
)

var (
	inInfoQname  = "reversionInfoIn"
	outInfoQname = "reversionInfoOut"
	client       http.Client
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

func initInfoQueue() *amqp.Channel {
	connection, err := amqp.Dial(config.Env.AmqpUri)
	failOnError(err, "Failed to connect to RabbitMQ")
	channel, err := connection.Channel()
	failOnError(err, "Failed to connect to RabbitMQ")
	_, err = channel.QueueDeclare(
		inInfoQname, // name
		false,       // durable
		false,       // delete when unused
		false,       // exclusive
		false,       // no-wait
		nil,         // arguments
	)
	failOnError(err, "Failed to connect to RabbitMQ")
	_, err = channel.QueueDeclare(
		outInfoQname, // name
		false,        // durable
		false,        // delete when unused
		false,        // exclusive
		false,        // no-wait
		nil,          // arguments
	)
	failOnError(err, "Failed to connect to RabbitMQ")
	return channel
}

func ProducerInfo() {

	mongoObj := mongodb.NewMongoClient(config.Env.MongoUri)
	outchan, wg := mongoObj.Get_Overdue_Revision_info()
	channel := initInfoQueue()
	// 发送任务
	var ct int
	for item := range outchan {
		ct += 1
		err := channel.Publish(
			"",          // exchange
			inInfoQname, // routing key
			false,       // mandatory
			false,       // immediate
			amqp.Publishing{
				ContentType: "application/octet-stream",
				Body:        []byte{},
				Headers: amqp.Table{
					"ID":             item.ID,
					"Title":          item.Title,
					"LastRevisionID": item.LastRevisionID,
				},
			})
		if err != nil {
			log.Println(err)
		}
	}
	wg.Wait()
	channel.Close()
	log.Info("上传队列结束，共上传任务数", ct)
}

func WorkerInfo() {
	channel := initInfoQueue()
	err := channel.Qos(
		10,    // prefetch count
		0,     // prefetch size
		false, // global
	)
	failOnError(err, "Failed to set QoS")

	msgs, err := channel.Consume(
		inInfoQname, // queue
		"",          // consumer
		false,       // auto-ack
		false,       // exclusive
		false,       // no-local
		false,       // no-wait
		nil,         // args
	)
	failOnError(err, "Failed to register a consumer")

	if config.Env.HttpProxy != "" {
		log.Info("use proxy:", config.Env.HttpProxy)
		var PTransport = &http.Transport{Proxy: http.ProxyFromEnvironment}
		client = http.Client{Transport: PTransport, Timeout: 120 * time.Second}
	} else {
		log.Info("no proxy")
		client = http.Client{Timeout: 120 * time.Second}
	}

	for d := range msgs {

		var rvstartid int64

		ID := d.Headers["ID"].(int64)
		LastRevisionID := d.Headers["LastRevisionID"].(int64)
		Title := d.Headers["Title"].(string)

		var ret_data reversion_pb.Reversion_Info_Job
		var allRevs []*reversion_pb.Reversion_Info
		for {
			status, lastrvendid, revs, err := get_reversions(Title, LastRevisionID, rvstartid)
			if err != nil {
				log.Info("请求错误:", Title, err)
				ret_data.State = false
				ret_data.Message = err.Error()
				break
			}
			if len(revs) > 0 {
				allRevs = append(allRevs, revs...)
				log.Infof("记录总数: %s - %d", Title, len(allRevs))
			}
			if status == 1 {
				log.Info("没有找到该条记录:", Title)
				ret_data.State = false
				ret_data.Message = "没有找到该条记录"
				break
			} else if status == 2 {
				log.Println("查询完毕:", Title)
				ret_data.State = true
				break
			} else if status == 3 {
				// loop
				log.Println("继续查询:", Title, lastrvendid)
				rvstartid = lastrvendid
			} else {
				log.Println("未知的status:", Title, status)
			}
		}
		ret_data.ReversionInfos = allRevs

		go func(qd amqp.Delivery) {
			log.Println("补全", ID)
			data, err := proto.Marshal(&ret_data)
			failOnError(err, "Marshal 失败")
			compressBody, err := compress.CompressData(data)
			failOnError(err, "compress 失败")
			for {
				err := channel.Publish(
					"",           // exchange
					outInfoQname, // routing key
					false,        // mandatory
					false,        // immediate
					amqp.Publishing{
						ContentType: "application/octet-stream",
						Body:        compressBody,
						Headers:     amqp.Table{"ID": ID},
					})
				if err != nil {
					log.Println("upload", err, " wait 1s ")
					<-time.After(1 * time.Second)
					continue
				}
				break
			}
			qd.Ack(false)
			log.Println("补全完成", ID)
		}(d)
	}

	forever := make(chan bool)
	log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
	<-forever
}

func ConsumerInfo() {

	mongoObj := mongodb.NewMongoClient(config.Env.MongoUri)
	channel := initInfoQueue()
	err := channel.Qos(
		10,    // prefetch count
		0,     // prefetch size
		false, // global
	)
	failOnError(err, "Failed to set QoS")

	msgs, err := channel.Consume(
		outInfoQname, // queue
		"",           // consumer
		false,        // auto-ack
		false,        // exclusive
		false,        // no-local
		false,        // no-wait
		nil,          // args
	)
	failOnError(err, "Failed to register a consumer")

	for d := range msgs {
		ID := d.Headers["ID"].(int64)
		log.Infof("start ID :%d", ID)
		data, err := compress.DecompressData(d.Body)
		if err != nil {
			log.Info("解压缩失败，跳过 article:", ID)
			continue
		}
		var reversion_info_job_obj reversion_pb.Reversion_Info_Job
		err = proto.Unmarshal(data, &reversion_info_job_obj)
		if err != nil {
			log.Info("Unmarshal 失败，跳过 article:", ID)
			continue
		}

		if !reversion_info_job_obj.State {
			// 采集失败的处理方法
			err := mongoObj.Mark_update_revision_info_fail(ID, reversion_info_job_obj.Message)
			if err != nil {
				log.Warn(err)
			}
		} else {
			// 成功就选择更新数据
			var reversionInfoList []*model.RevisionInfoObj
			for _, item := range reversion_info_job_obj.ReversionInfos {
				timestamp := item.Timestamp.AsTime()
				reversionInfoList = append(reversionInfoList, &model.RevisionInfoObj{
					Revid:     item.RevID,
					ArticleID: item.ArticleID,
					Comment:   item.Comment,
					ParentID:  item.ParentID,
					User:      item.User,
					Timestamp: timestamp,
				})
			}
			// 交个数据库
			if len(reversionInfoList) > 0 {
				mongoObj.Bulk_update_revision_info(ID, reversionInfoList)
			} else {
				err := mongoObj.Mark_update_revision_info_fail(ID, "结果长度为0")
				if err != nil {
					log.Warn(err)
				}
			}
		}
		d.Ack(true)
	}

	forever := make(chan bool)
	log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
	<-forever
}

func replace_wiki_name(article_name string) string {
	k_m := map[string]string{
		" ": "_",
		// ",": "%2C",
		// "/": "%2F",
		// "$": "%24",
		// "?": "%3F",
		// "&": "%26",
		// "%": "%25",
	}
	for k, v := range k_m {
		article_name = strings.Replace(article_name, k, v, -1)
	}
	return article_name
}

// status 1,not found name; 2, over 3, loop
// lastrevid
// revionse
func formatResponse(responseBody []byte) (status int, lastrvendid int64, revs []*reversion_pb.Reversion_Info) {

	var repData map[string]interface{}
	json.Unmarshal(responseBody, &repData)

	// cont := repData["continue"]
	// 没有query的情况
	if _, found := repData["query"].(map[string]interface{})["pages"]; !found {
		log.Info("no page found")
		status = 1
		return
	}
	query := repData["query"].(map[string]interface{})["pages"].(map[string]interface{})
	var mypage map[string]interface{}
	for key, page := range query {
		if key == "-1" {
			status = 1
			return
		}
		mypage = page.(map[string]interface{})
		break
	}

	// 补丁
	if mypage["revisions"] == nil {
		status = 1
		return
	}
	page_id := int64(mypage["pageid"].(float64))
	for _, itemdata := range mypage["revisions"].([]interface{}) {
		item := itemdata.(map[string]interface{})
		revTimestamp, _ := time.Parse("2006-01-02T15:04:05Z", item["timestamp"].(string))
		timetamp := timestamppb.New(revTimestamp)
		reversion_info := &reversion_pb.Reversion_Info{
			ArticleID: page_id,
			RevID:     int64(item["revid"].(float64)),
			ParentID:  int64(item["parentid"].(float64)),
			Timestamp: timetamp,
		}

		if comment, found := item["comment"]; found {
			reversion_info.Comment = comment.(string)
		}
		if user, found := item["user"]; found {
			reversion_info.User = user.(string)
		}
		revs = append(revs, reversion_info)
	}
	lastrvendid = revs[len(revs)-1].RevID
	// if 'missing' in page and 'revisions' not in page:
	if mypage["missing"] != nil && mypage["revisions"] == nil {
		status = 1
	} else if repData["batchcomplete"] != nil {
		status = 2
	} else if repData["continue"] != nil {
		status = 3
	} else {
		log.Warn("not get request work status")
		log.Warn("%+v", repData)
	}
	return
}

// status 1,not found name; 2, over 3, loop
// lastrevid
// revionse
func get_reversions(article_name string, rvendid int64, rvstartid int64) (status int, lastrvendid int64, revs []*reversion_pb.Reversion_Info, err error) {
	// log.Println("开始任务:", article_name, rvendid)
	article_name = replace_wiki_name(article_name)

	base_url := "https://en.wikipedia.org/w/api.php"
	req, _ := http.NewRequest("GET", base_url, nil)
	q := req.URL.Query()
	q.Add("action", "query")
	q.Add("format", "json")
	q.Add("prop", "revisions")
	q.Add("rvlimit", "max")
	q.Add("titles", replace_wiki_name(article_name))
	if rvendid != 0 {
		q.Add("rvendid", strconv.Itoa(int(rvendid)))
	}
	if rvstartid != 0 {
		q.Add("rvstartid", strconv.Itoa(int(rvstartid)))
	}
	req.URL.RawQuery = q.Encode()

	var (
		resp *http.Response
	)
	log.Info(req.URL.String())
	for i := 0; i <= 3; i++ {
		resp, err = client.Do(req)
		if err != nil {
			if i == 3 {
				return
			}
			log.Println("请求失败，正在重试：", i, err)
			time.Sleep(1 * time.Second)
			continue
		}
		if resp.StatusCode < 200 || resp.StatusCode >= 300 {
			time.Sleep(1 * time.Second)
			continue
		}

		break
	}
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		err = fmt.Errorf("statu no 200:%d", resp.StatusCode)
		time.Sleep(1 * time.Second)
		return
	}

	content, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return
	}
	status, lastrvendid, revs = formatResponse(content)
	return
}
